package com.my.basic.flow.backpressure;

import java.util.concurrent.Flow;

/**
 * @author: marshal
 * @date: 2022-07-27 16:10
 */
public class MySubscriber implements Flow.Subscriber<Integer> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println(Thread.currentThread().getName() + " | subscription, beginning request data");
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(Integer item) {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(Thread.currentThread().getName() + " | subscriber beginning receive data : " + item);
        if (item == 0) {
            this.subscription.request(5);
        }
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("receive data error : " + throwable.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("subscribe complete");
    }
}
